package com.chencong.online.driver.dwd;

import com.chencong.online.bean.LogFailWarningBean;
import com.chencong.online.bean.LoginEventBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.LoginEventMapFunc;
import com.chencong.online.function.LoginFailPattenFunc;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.net.URL;
import java.time.Duration;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdLoginFailAnalysisDriver
 * @description:利用cep实时监控在2s内连续登陆失败的用户
 * @author: chencong
 * @create: 2021-12-31 10:29
 **/
public class DwdLoginFailAnalysisDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);

        //1、source
        URL resource = DwdLoginFailAnalysisDriver.class.getResource("/LoginLog.csv");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());
//        inputDS.print();
        //2、transform
        //将数据转换为javabean并提取时间戳
        SingleOutputStreamOperator<LoginEventBean> loginEventBeanDS = inputDS
                .map(new LoginEventMapFunc())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<LoginEventBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((SerializableTimestampAssigner<LoginEventBean>) (element, recordTimestamp) -> element.getTimestamp() * 1000L));

        //todo 1 定义一个匹配规则
        //firstFail -> secondFail ,within 2s
        Pattern<LoginEventBean, LoginEventBean> loginEventPatten = Pattern
                .<LoginEventBean>begin("firstFail").where(new SimpleCondition<LoginEventBean>() {
                    @Override
                    public boolean filter(LoginEventBean value) throws Exception {
                        return "fail".equals(value.getLoginState());
                    }
                })
                .next("secondFail").where(new SimpleCondition<LoginEventBean>() {
                    @Override
                    public boolean filter(LoginEventBean value) throws Exception {
                        return "fail".equals(value.getLoginState());
                    }
                })
                .within(Time.seconds(3));

        //todo 2 将匹配模式应用到数据流上，得到一个pattern stream
        PatternStream<LoginEventBean> patternDS = CEP.pattern(loginEventBeanDS.keyBy(data -> data.getUserId()), loginEventPatten);

        //todo 3 检出符合匹配条件的复杂事件，进行转换处理，得到报警信息
        SingleOutputStreamOperator<LogFailWarningBean> warningDS = patternDS.select(new LoginFailPattenFunc());
        warningDS.print("报警信息！！");

        env.execute("DwdLoginFailAnalysisDriver");
    }
}
